package kvraft

import (
	"6.824/labgob"
	"6.824/labrpc"
	"6.824/raft"
	"log"
	"sync"
	"sync/atomic"
	"time"
)

const Debug = true

func DPrintf(format string, a ...interface{}) (n int, err error) {
	if Debug {
		log.Printf(format, a...)
	}
	return
}

type Op struct {
	// Your definitions here.
	// Field names must start with capital letters,
	// otherwise RPC will break.

	// Command
	Operate string
	Key     string
	Value   string
	// Linear
	ClientID  int64
	RequestID int
}

type KVServer struct {
	mu           sync.Mutex
	me           int
	rf           *raft.Raft
	applyCh      chan raft.ApplyMsg
	dead         int32 // set by Kill()
	maxraftstate int   // snapshot if log grows this big
	// Your definitions here.
	stateMachine StateMachine
}

// 使用start 发送信息
func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	// Your code here.
	if result, yes := kv.stateMachine.isRequested(args.ClientID, args.RequestID); yes {
		reply.Err = OK
		reply.Value = result
		return
	}
	command := Op{
		Operate:   GET,
		Key:       args.Key,
		ClientID:  args.ClientID,
		RequestID: args.RequestID,
	}
	_, _, isLeader := kv.rf.Start(command)
	if !isLeader {
		DPrintf("wrong Leader %v", kv.me)
		reply.Err = ErrWrongLeader
	} else {
		//DPrintf("receive query %v", args)
		kv.stateMachine.CheckAndCreateRequest(args.ClientID)
		select {
		case reply.Value = <-kv.stateMachine.Client[args.ClientID].ch:
			reply.Err = OK
		case <-time.After(time.Second):
			reply.Err = ErrWrongLeader
		}
	}
}

func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	// Your code here.
	// 先查询请求是否是重复请求
	if _, yes := kv.stateMachine.isRequested(args.ClientID, args.RequestID); yes {
		reply.Err = OK
		return
	}
	command := Op{
		Operate:   args.Op,
		Key:       args.Key,
		Value:     args.Value,
		ClientID:  args.ClientID,
		RequestID: args.RequestID,
	}
	_, _, isLeader := kv.rf.Start(command)
	if !isLeader {
		reply.Err = ErrWrongLeader
	} else {
		//DPrintf("receive query %v", args)
		kv.stateMachine.CheckAndCreateRequest(args.ClientID)
		ch := kv.stateMachine.Client[args.ClientID].ch
		select {
		case <-ch:
			reply.Err = OK
		case <-time.After(time.Second):
			DPrintf("raft has not apply the command in 1s")
			reply.Err = ErrWrongLeader
		}
	}
}

// the tester calls Kill() when a KVServer instance won't
// be needed again. for your convenience, we supply
// code to set rf.dead (without needing a lock),
// and a killed() method to test rf.dead in
// long-running loops. you can also add your own
// code to Kill(). you're not required to do anything
// about this, but it may be convenient (for example)
// to suppress debug output from a Kill()ed instance.
func (kv *KVServer) Kill() {
	atomic.StoreInt32(&kv.dead, 1)
	kv.rf.Kill()
	// Your code here, if desired.
}

func (kv *KVServer) killed() bool {
	z := atomic.LoadInt32(&kv.dead)
	return z == 1
}

// servers[] contains the ports of the set of
// servers that will cooperate via Raft to
// form the fault-tolerant key/value service.
// me is the index of the current server in servers[].
// the k/v server should store snapshots through the underlying Raft
// implementation, which should call persister.SaveStateAndSnapshot() to
// atomically save the Raft state along with the snapshot.
// the k/v server should snapshot when Raft's saved state exceeds maxraftstate bytes,
// in order to allow Raft to garbage-collect its log. if maxraftstate is -1,
// you don't need to snapshot.
// StartKVServer() must return quickly, so it should start goroutines
// for any long-running work.
func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
	// call labgob.Register on structures you want
	// Go's RPC library to marshall/unmarshall.
	labgob.Register(Op{})
	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate
	// You may need initialization code here.

	kv.applyCh = make(chan raft.ApplyMsg)
	kv.rf = raft.Make(servers, me, persister, kv.applyCh)
	kv.stateMachine.KV = make(map[string]string)
	kv.stateMachine.Client = make(map[int64]Request)
	// You may need initialization code here.
	go kv.dealWithApply()
	return kv
}

func (kv *KVServer) dealWithApply() {
	for !kv.killed() {
		apply := <-kv.applyCh
		DPrintf("APPLY command %v", apply.Command)
		if apply.CommandValid {
			op, ok := (apply.Command).(Op)
			if ok {
				// 更新状态机，还应该发送消息通知等待的线程检查状态机
				// apply去重
				if rep, yes := kv.stateMachine.isRequested(op.ClientID, op.RequestID); yes {
					kv.stateMachine.Client[op.ClientID].ch <- rep
				} else {
					if op.Operate == PUT {
						kv.stateMachine.put(op.Key, op.Value)
						kv.stateMachine.SaveResult(op.ClientID, op.RequestID, "")
						kv.stateMachine.Client[op.ClientID].ch <- ""
					} else if op.Operate == APPEND {
						kv.stateMachine.append(op.Key, op.Value)
						kv.stateMachine.SaveResult(op.ClientID, op.RequestID, "")
						kv.stateMachine.Client[op.ClientID].ch <- ""
					} else {
						result := kv.stateMachine.get(op.Key)
						kv.stateMachine.Client[op.ClientID].ch <- result
					}
				}
			}
		}
	}
}
